package com.chenjj.bigdata.flink.cdc;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.debezium.DebeziumDeserializationSchema;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

public class MysqlCdc {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(new Configuration());

        // 使用MySQLSource创建数据源
        // 同时指定StringDebeziumDeserializationSchema，将CDC转换为String类型输出
        DebeziumDeserializationSchema schema = new StringDebeziumDeserializationSchema();
        SourceFunction sourceFunction = MySQLSource.builder().hostname("8.134.99.21").port(3306)
                .databaseList("poc_test").username("root").password("root200")
                .deserializer(schema).build();

        // 单并行度打印，避免输出乱序
        env.addSource(sourceFunction).print().setParallelism(1);
        env.execute();
    }
}
